package com.linshidream.intrachat.simplenio.pipeline.boots;

import com.linshidream.intrachat.simplenio.pipeline.context.MiniChannel;
import com.linshidream.intrachat.simplenio.pipeline.hearbeat.ScheduledTask;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created on 2025/8/17 22:00
 *
 * @author linshidream
 * @version 1.0.0
 * @description
 */
public class MiniWorker implements Runnable {

    private final BlockingQueue<MiniChannel> chs = new LinkedBlockingDeque<>();


    /**
     * 优先级队列，会按照 ScheduledTask.deadline 从小到大排序
     */
    private final PriorityQueue<ScheduledTask> scheduledTasks = new PriorityQueue<>(Comparator.comparingLong(ScheduledTask::getDeadline));

    /**
     * 一个 PacketWorker 对象持有一个监听器
     */
    private final Selector selector;

    public Selector nextSelector() {
        return selector;
    }

    /**
     * 每个socket对象持有一个上下文连接
     */
    private final Map<SocketChannel, MiniChannel> contextMap;


    public MiniWorker() throws IOException {
        this.selector = Selector.open();
        // 私有 Map，无需线程安全
        this.contextMap = new HashMap<>();
    }

    public void register(MiniChannel ch) {
        chs.offer(ch);
        // 这里一定要唤醒 run里面就开始取 SocketChannel 去注册
        selector.wakeup();
    }


    /**
     * 添加定时任务
     *
     * @param task
     * @param delayMillis
     */
    public void schedule(Runnable task, long delayMillis) {
        scheduledTasks.add(new ScheduledTask(System.currentTimeMillis() + delayMillis, task));

        // 唤醒一下，让 select 的超时重新计算
        selector.wakeup();
    }

    public void fireIdleCheck(MiniChannel channel, long initialDelayMillis, long periodMillis) {
        schedule(() -> {
            if (!channel.isActive()) {
                // 通道已经关闭，停止递归调度
                return;
            }
            System.out.println("[ScheduledTask] 空闲检测...");
            // 执行空闲检查
            channel.pipeline().fireIdleCheck();

            fireIdleCheck(channel, periodMillis, periodMillis);
        }, initialDelayMillis);
    }


    /**
     * 计算最近一个定时任务的 delay
     */
    private long getNextScheduledTaskDelay() {
        ScheduledTask next = scheduledTasks.peek();
        if (next == null) {
            // 默认 1s 触发一次空闲事件检测
            return 1000;
        }
        long delay = next.getDeadline() - System.currentTimeMillis();
        return Math.max(delay, 0);
    }


    /**
     * 执行到期的定时任务
     */
    private void runScheduledTasks() {
        long now = System.currentTimeMillis();
        while (!scheduledTasks.isEmpty() && scheduledTasks.peek().getDeadline() <= now) {
            ScheduledTask task = scheduledTasks.poll();
            if (task != null) {
                task.getTask().run();
            }
        }
    }


    @Override
    public void run() {
        try {
            while (true) {
                // selector.select();

                // 1、会把所有的 SocketChannel 注册到 selector，直到注册完成
                MiniChannel ch;
                while ((ch = chs.poll()) != null) {
                    SocketChannel sc = ch.socketChannel();
                    sc.configureBlocking(false);
                    sc.register(selector, SelectionKey.OP_READ);

                    // 建立连接
                    contextMap.put(sc, ch);
                }

                // 2. 计算最近定时任务的 delay，带超时 select，
                // 比如任务 5 秒后执行，那么就5秒后唤醒
                // 如果  selector.wakeup(); 会立刻唤醒
                long nextDelay = getNextScheduledTaskDelay();
                int readyChannels = selector.select(nextDelay);


                // 3、如果有 IO 事件，处理它
                if (readyChannels > 0) {
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();

                        if (selectionKey.isValid() && selectionKey.isReadable()) {
                            SocketChannel client = (SocketChannel) selectionKey.channel();
                            MiniChannel sc = contextMap.get(client);
                            // 把通道交给责任链 支持累积缓冲区
                            MiniReaderWriter.doRead(sc);
                        }

                        if (selectionKey.isValid() && selectionKey.isWritable()) {
                            SocketChannel client = (SocketChannel) selectionKey.channel();
                            MiniChannel sc = contextMap.get(client);

                            // 写监听 上次未写完的数据可以继续写（socket 缓冲区满了）
                            MiniReaderWriter.doWrite(sc);
                        }
                    }
                }

                // 4. 执行到期的定时任务（心跳检测/超时）
                runScheduledTasks();
            }
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("MiniWorker 线程执行异常," + Thread.currentThread().getName());
        }
    }
}
